/*
 * Copyright 2011 Dominik Pretzsch <dominik.pretzsch at gmail.com>.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package de.blacksheep.jmoteextract.thread;

import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * <code>ThreadDispatcher</code> provides access to a thread pool. <br>
 * The size of the pool can be specified. Any callable <br>
 * can be submitted whenever the dispatcher is not <br>
 * busy (numThreads is less maxThreads).
 * 
 * @author Dominik Pretzsch <dominik.pretzsch at gmail.com>
 */
public class ThreadDispatcher {
    
    private final ExecutorService threadPool;
    private ConcurrentHashMap<Integer, Future<Boolean>> callableFutureMap;
    private int maxThreads;
    
    private static final Logger LOGGER =  Logger.getLogger(ThreadDispatcher.class.getName());

    public ThreadDispatcher(int maxThreads) {
        this.maxThreads = (maxThreads > 0) ? maxThreads : 1;
        callableFutureMap = new ConcurrentHashMap<Integer, Future<Boolean>>();
        threadPool = Executors.newFixedThreadPool(maxThreads);
    }  
    
    public ThreadDispatcher() {
        this.maxThreads = 0;
        callableFutureMap = new ConcurrentHashMap<Integer, Future<Boolean>>();
        threadPool = Executors.newCachedThreadPool();
    }
    
    /**
     * Submits a callable to the {@see threadPool}. <br>
     * Returns either the hash value of the <code>Callable</code> <br> 
     * or <code>0</code> if <code>ThreadDispatcher</code> is busy. <br>
     * 
     * @param callable
     * @return int  Hash value of Callable or 0 if busy.
     */
    public int submit(Callable<Boolean> callable) {
        int hash = callable.hashCode();
        if (!isBusy()) {            
            LOGGER.log(Level.FINE, "Submitting Callable {0}...", hash);            
            Future<Boolean> future = threadPool.submit(callable);
            callableFutureMap.put(hash, future);            
            return hash;
        }
        return 0;
    }
    
    /**
     * Returns whether all threads in {@see #callableFutureMap} <br>
     * have finished.
     * 
     * @return true     All threads have finished. <br>
     * @return false    Not all threads have finished, yet.
     */
    public boolean allDone() {
        boolean allDone = true;
        Set<Entry<Integer, Future<Boolean>>> callables = callableFutureMap.entrySet();
        for (Entry<Integer, Future<Boolean>> entry : callables) {
            int hash = entry.getKey();
            Future result = entry.getValue();
            if (result.isDone()) {
                LOGGER.log(Level.FINE, "Removing Callable {0}...", hash);                
                callableFutureMap.remove(hash);
            } else {
                allDone = false;
            }
        }
        return allDone;
    }
    
    /**
     * Returns whether <code>numThreads == maxThreads</code>.
     * 
     * @return true     Dispatcher is busy.<br>
     * @return false    Dispatcher is not busy.
     */
    public boolean isBusy() {
        return (maxThreads == 0) ? !(callableFutureMap.size() < maxThreads) : false;
    }
}
